package lab8;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.io.Text;

public class UidMain {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

		// 源文件输入路径，就是sogou.txt文件的路径，首先要将其上传到HDFS
		// 假设上传到HDFS文件系统的路径为hdfs：//192.168.2.131:9000/test/sogou.txt
		String inpath = "hdfs://192.168.2.131:9000/test/sogou.txt";
		// 经过MapReduce数据处理之后结果的输出文件路径，注意该路径不能事先存在
		// 假设设置的路径为hdfs://192.168.2.131/test/sogououtput
		String outpath = "hdfs://192.168.2.131:9000/test/sogououtput";

		// 创建MapReduce的job对象，并设置job的名称
		Job job = Job.getInstance(new Configuration(), UidMain.class.getSimpleName());
		// 设置job作业运行时的程序入口主类WordCount
		job.setJarByClass(UidMain.class);

		// 设置作业运行时map阶段的类对象
		job.setMapperClass(UidMapper.class);
		// 设置作业运行时reduce阶段的类对象
		job.setReducerClass(UidReducer.class);

		// 设置job作业输入、输出格式为文本格式
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		// 设置map方法执行中间结果输出的key类型为Text
		job.setMapOutputKeyClass(Text.class);
		// 设置map方法执行中间结果输出的value类型
		job.setMapOutputValueClass(IntWritable.class);

		// 设置job输出的key类型为Text
		job.setOutputKeyClass(Text.class);
		// 设置job输出的value类型为IntWritable
		job.setOutputValueClass(IntWritable.class);

		// 设置输入文件的路径
		FileInputFormat.addInputPath(job, new Path(inpath));
		// 设置计算结果的输出路径
		FileOutputFormat.setOutputPath(job, new Path(outpath));

		// 提交运行job
		int num = job.waitForCompletion(true) ? 0 : 1;
		// 根据job执行返回的结果退出程序
		System.exit(num);
	}
}